package com.zy.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.bean.TableProcess;
import com.zy.gmall.realtime.common.GmallConfig;
import com.zy.gmall.realtime.util.DimUtil;
import com.zy.gmall.realtime.util.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

public class MySinkRichFunction extends RichSinkFunction<JSONObject> {

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取对业务数据库维度表的操作的类型
        String type = value.getString("type");
        //获取当前操作的这条维度数据对应的配置表信息
        TableProcess tableProcess = value.getObject("table_process", TableProcess.class);
        //获取操作的hbase中表名
        String table = tableProcess.getSinkTable();
        //获取rowkey
        String sinkRowKey = tableProcess.getSinkRowKey();
        String rowKey = value.getString(sinkRowKey);

        if ("delete".equals(type)){
            //根据rowKey删除数据
            System.out.println("删除!!!!!!!!!!!!!!!!");
            HbaseUtil.delData(conn, GmallConfig.HBASE_NAMESPACE,table,rowKey);
        }else {
            System.out.println("新增！！！！！！！！！！！！！！！");
            String family = tableProcess.getSinkFamily();
            String[] columns = tableProcess.getSinkColumns().split(",");
            String[] values = new String[columns.length];
            //获取列名和值
            for (int i = 0; i < values.length; i++) {
                String column = columns[i];
                values[i] = value.getString(column);
            }
            HbaseUtil.addData(conn,GmallConfig.HBASE_NAMESPACE,rowKey,table,family,columns,values);
        }
        if ("update".equals(type) || "delete".equals(type)){
            DimUtil.delCached(table,rowKey);
        }
    }

    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getConnect();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbase(conn);
    }

}
